package com.bw.util;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.tools.javac.util.ListBuffer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.kafka.common.protocol.types.Field;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

public class SourceHbaseUntil {

    public static String columnFamily="data";

    public static String columnName="tagname";

    public static String sc(String tableName, String rowkey) throws Exception {
        Connection conn = HBaseUtil.getConnection(ConfigUtil.getZookeeperQuorum(), ConfigUtil.getZookeeperClientPort());
        Table table = conn.getTable(TableName.valueOf(tableName));

        Get get = new Get(Bytes.toBytes(rowkey));
        get.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(columnName));
        Result result = table.get(get);

        String tagname=null;
        for (Cell cell : result.listCells()) {
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            tagname=value;
        }

        table.close();
        conn.close();

        return tagname;
    }

    public static void scc(String tableName) throws Exception {
        //创建Hbase连接
        Connection conn = HBaseUtil.getConnection(ConfigUtil.getZookeeperQuorum(), ConfigUtil.getZookeeperClientPort());
        Table table = conn.getTable(TableName.valueOf(tableName));
        //
        Scan scan = new Scan();

        ResultScanner results = table.getScanner(scan);
        ObjectMapper objectMapper = new ObjectMapper(); // 创建Jackson对象映射器

        for (Result re : results) {
            Cell[] cells = re.rawCells();
            HashMap<String, String> rowData = new HashMap<>();
            for (Cell cell : cells) {
                rowData.put(Bytes.toString(CellUtil.cloneQualifier(cell)),Bytes.toString(CellUtil.cloneValue(cell)));
            }
            String jsonString = objectMapper.writeValueAsString(rowData);// 将Map转换为JSON字符串
            System.out.println(jsonString);
        }

        table.close();
        conn.close();
    }

    public static void main(String[] args) throws Exception {
        String tableName = "GMALL2022_REALTIME:dim_base_house";
        String rowkey="151010";
        String ColumnFamily="data";
        List<String> Qualifier= Arrays.asList("id","tagcode","tagname");

        try {
//            SourceHbaseUntil.sc(tableName,ColumnFamily,Qualifier);
            String s = SourceHbaseUntil.sc(tableName, rowkey);
            System.out.println(s);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
